from __future__ import annotations

"""
信号异步处理器（SignalWorker）
-----------------------------
职责：
- 异步消费 TradingView Webhook 入队的信号，按订阅与风控规则执行下单
- 进行速率限制、白名单校验、名义金额与规则归一化计算
- 记录执行订单与审计日志，更新信号最终状态

流程概要：
1. 拉取信号与策略，检查策略是否激活
2. 查询所有订阅用户，过滤未启用/全局暂停用户
3. 为每个订阅计算名义金额与下单数量，应用交易规则
4. 调 hummingbot-api 下单并落库执行结果
5. 汇总成功/失败并设置信号状态（filled/partial/error）
"""

import asyncio
import logging
from decimal import Decimal
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from ..config import get_settings
from ..db import SessionLocal
from ..models import (
    AuditLog,
    ExecOrder,
    MarketType,
    OrderStatus,
    PositionAction,
    Script,
    ScriptSignal,
    ScriptStatus,
    ScriptSubscription,
    SignalStatus,
    TradeSide,
    User,
)
from .hb_client import HBClientError, hb_client
from .rate_limit import user_rate_limiter
from .risk import (
    RiskError,
    enforce_max_notional,
    enforce_whitelists,
    floor_to_step,
    get_amount_step,
    get_min_base_amount,
    get_reference_price,
)
from .utils import default_account_name, now_utc, translate_error_message

logger = logging.getLogger(__name__)
settings = get_settings()


class SignalWorker:
    def __init__(self, concurrency: int = 2) -> None:
        self._queue: asyncio.Queue[int] = asyncio.Queue()
        self._tasks: List[asyncio.Task] = []
        self._concurrency = concurrency
        self._stop_event = asyncio.Event()

    @staticmethod
    def _connector_matches_market(connector_name: str, market_type: MarketType) -> bool:
        name = (connector_name or "").lower()
        if not name:
            return False
        is_perp = any(keyword in name for keyword in ("perp", "swap", "future", "futures"))
        if market_type == MarketType.perpetual:
            return is_perp
        return not is_perp

    @staticmethod
    def _to_decimal(value: Any) -> Optional[Decimal]:
        if value is None:
            return None
        try:
            candidate = Decimal(str(value))
        except Exception:  # noqa: BLE001
            return None
        if candidate <= 0:
            return None
        return candidate

    @classmethod
    def _compute_amount_and_price(
        cls,
        *,
        market_type: MarketType,
        order_type: str,
        trade_type: str,
        position_action: str,
        notional: Decimal,
        explicit_price: Optional[Decimal],
        rule: Dict[str, Any],
        best_bid: Optional[Decimal],
        best_ask: Optional[Decimal],
        slippage_bps: int,
    ) -> Tuple[Decimal, Optional[Decimal]]:
        step = get_amount_step(rule)
        min_base = get_min_base_amount(rule)
        fallback_price = explicit_price if explicit_price and explicit_price > 0 else None
        if fallback_price is None:
            ref_price = get_reference_price(rule)
            if ref_price > 0:
                fallback_price = ref_price

        slippage = Decimal("0")
        if slippage_bps and slippage_bps > 0:
            slippage = Decimal(str(slippage_bps)) / Decimal("10000")
        if slippage < 0:
            slippage = Decimal("0")

        normalized_order_type = order_type.upper()
        normalized_trade_type = trade_type.upper()
        normalized_position_action = position_action.upper()
        # 保留 market_type / position_action，便于后续扩展平仓限制等逻辑
        _ = (market_type, normalized_position_action)

        step_applied = step if step and step > 0 else Decimal("0.00000001")

        if normalized_order_type == "LIMIT":
            if explicit_price is None or explicit_price <= 0:
                raise RiskError("ERR_PRICE_REQUIRED", "限价单缺少价格")
            raw_amount = notional / explicit_price
            floored = floor_to_step(raw_amount, step_applied)
            amount = floored
            if min_base > 0 and amount < min_base:
                amount = min_base
            if amount <= 0:
                raise RiskError("ERR_AMOUNT", "计算得到的下单数量无效")
            return amount, explicit_price

        # 市价
        price_for_amount: Optional[Decimal] = None
        if normalized_trade_type == "BUY":
            price_for_amount = best_ask
            factor = Decimal("1") + slippage
        else:
            price_for_amount = best_bid
            factor = Decimal("1") - slippage

        if price_for_amount is None or price_for_amount <= 0:
            # 使用兜底参考价（规则/显式价格/均价）
            if fallback_price is None and best_bid and best_ask:
                fallback_price = (best_bid + best_ask) / Decimal("2")
            price_for_amount = fallback_price

        if price_for_amount is None or price_for_amount <= 0:
            raise RiskError("ERR_MARKET_PRICE", "缺少市价参考，无法计算下单数量")
        if factor <= 0:
            raise RiskError("ERR_SLIPPAGE", "滑点参数导致成交价格无效")

        adjusted_price = price_for_amount * factor
        raw_amount = notional / adjusted_price
        floored = floor_to_step(raw_amount, step_applied)
        amount = floored
        if min_base > 0 and amount < min_base:
            amount = min_base

        if amount <= 0:
            raise RiskError("ERR_AMOUNT", "计算得到的下单数量无效")

        # 平仓额外约束：若未来需要，可在此处查询仓位并限制 amount
        return amount, None

    @staticmethod
    def _extract_book_price(side_data: Any) -> Optional[Decimal]:
        if side_data is None:
            return None
        if isinstance(side_data, dict):
            for key in ("price", "p", "best_price", "value"):
                if key in side_data:
                    price = SignalWorker._to_decimal(side_data.get(key))
                    if price is not None:
                        return price
            levels = side_data.get("levels")
            if isinstance(levels, list):
                return SignalWorker._extract_book_price(levels)
        if isinstance(side_data, (list, tuple)):
            if not side_data:
                return None
            first = side_data[0]
            if isinstance(first, (list, tuple)):
                if not first:
                    return None
                return SignalWorker._to_decimal(first[0])
            if isinstance(first, dict):
                return SignalWorker._extract_book_price(first)
            return SignalWorker._to_decimal(first)
        return SignalWorker._to_decimal(side_data)

    @classmethod
    def _parse_order_book(cls, raw: Any) -> Tuple[Optional[Decimal], Optional[Decimal]]:
        data = raw
        if isinstance(data, dict) and "data" in data and isinstance(data["data"], dict):
            data = data["data"]
        if isinstance(data, list):
            if data and isinstance(data[0], dict):
                data = data[0]
            else:
                bids = cls._extract_book_price(data)
                return bids, None

        bids = None
        asks = None
        if isinstance(data, dict):
            bids = cls._extract_book_price(
                data.get("bids")
                or data.get("bid")
                or data.get("buy")
                or data.get("levels_bid")
            )
            asks = cls._extract_book_price(
                data.get("asks")
                or data.get("ask")
                or data.get("sell")
                or data.get("levels_ask")
            )
            if bids is None:
                bids = cls._to_decimal(
                    data.get("best_bid")
                    or data.get("best_bid_price")
                    or data.get("bid_price")
                )
            if asks is None:
                asks = cls._to_decimal(
                    data.get("best_ask")
                    or data.get("best_ask_price")
                    or data.get("ask_price")
                )
        return bids, asks

    async def _get_best_quotes(
        self,
        connector_name: str,
        symbol: str,
        cache: Dict[Tuple[str, str], Tuple[Optional[Decimal], Optional[Decimal]]],
    ) -> Tuple[Optional[Decimal], Optional[Decimal]]:
        key = (connector_name.lower(), symbol.upper())
        if key in cache:
            return cache[key]
        try:
            order_book = await hb_client.get_order_book(connector_name, symbol, depth=1)
        except HBClientError as exc:
            raise RiskError("ERR_MARKET_DATA", "获取盘口价格失败") from exc
        bid, ask = self._parse_order_book(order_book)
        cache[key] = (bid, ask)
        return bid, ask

    async def start(self) -> None:
        if self._tasks:
            return
        for _ in range(self._concurrency):
            task = asyncio.create_task(self._worker_loop())
            self._tasks.append(task)
        logger.info("Signal worker started with %s consumers", self._concurrency)

    async def stop(self) -> None:
        self._stop_event.set()
        for task in self._tasks:
            task.cancel()
        await asyncio.gather(*self._tasks, return_exceptions=True)
        self._tasks.clear()
        await hb_client.close()
        logger.info("Signal worker stopped")

    async def submit(self, signal_id: int) -> None:
        await self._queue.put(signal_id)

    async def _worker_loop(self) -> None:
        while not self._stop_event.is_set():
            try:
                signal_id = await self._queue.get()
            except asyncio.CancelledError:
                break
            try:
                await self._process_signal(signal_id)
            except Exception as exc:  # noqa: BLE001
                logger.exception("Error processing signal %s: %s", signal_id, exc)
            finally:
                self._queue.task_done()

    async def _process_signal(self, signal_id: int) -> None:
        async with SessionLocal() as session:
            signal = await session.get(ScriptSignal, signal_id)
            if signal is None:
                logger.warning("Signal %s not found", signal_id)
                return

            script = await session.get(Script, signal.script_id)
            if script is None:
                await self._mark_signal(
                    session,
                    signal,
                    SignalStatus.error,
                    translate_error_message(code="script_missing"),
                )
                return

            await self._mark_signal(session, signal, SignalStatus.processing)

            if script.status != ScriptStatus.active:
                await self._mark_signal(
                    session,
                    signal,
                    SignalStatus.error,
                    translate_error_message(code="script_paused"),
                )
                return

            subscriptions = await self._fetch_subscriptions(session, script.id)

            if not subscriptions:
                await self._mark_signal(
                    session,
                    signal,
                    SignalStatus.error,
                    translate_error_message(code="no_subscribers"),
                )
                return

            payload = signal.payload_json
            trade_type = str(payload.get("trade_type", "BUY")).upper()
            order_type = str(payload.get("order_type", "MARKET")).upper()
            position_action = str(payload.get("position_action", "OPEN")).upper()
            symbol = str(payload.get("trading_pair") or payload.get("symbol") or script.symbol or "")
            price = payload.get("price")

            rules_cache: Dict[tuple[str, str], Dict[str, Any]] = {}
            order_book_cache: Dict[Tuple[str, str], Tuple[Optional[Decimal], Optional[Decimal]]] = {}
            success = 0
            failures = 0

            for sub, user in subscriptions:
                if not sub.enabled or not user.tv_enabled or user.global_pause:
                    continue

                connector_name = (sub.connector_name or "").strip()
                account_name = (sub.account_name or "").strip() or default_account_name(user.id, user.email)

                if not connector_name:
                    await self._record_exec_order(
                        session=session,
                        signal=signal,
                        user=user,
                        account_name=account_name,
                        connector="",
                        symbol=symbol,
                        side=trade_type,
                        order_type=order_type,
                        position_action=position_action,
                        qty=Decimal("0"),
                        price=price,
                        status=OrderStatus.rejected,
                        error=translate_error_message(code="missing_connector"),
                    )
                    failures += 1
                    continue

                if not self._connector_matches_market(connector_name, script.market_type):
                    await self._record_exec_order(
                        session=session,
                        signal=signal,
                        user=user,
                        account_name=account_name,
                        connector=connector_name,
                        symbol=symbol,
                        side=trade_type,
                        order_type=order_type,
                        position_action=position_action,
                        qty=Decimal("0"),
                        price=price,
                        status=OrderStatus.rejected,
                        error=translate_error_message(code="connector_mismatch"),
                    )
                    failures += 1
                    continue

                allowed = await user_rate_limiter.allow(f"user:{user.id}", get_settings().user_rate_per_min)
                if not allowed:
                    await self._record_exec_order(
                        session=session,
                        signal=signal,
                        user=user,
                        account_name=account_name,
                        connector=connector_name,
                        symbol=symbol,
                        side=trade_type,
                        order_type=order_type,
                        position_action=position_action,
                        qty=Decimal("0"),
                        price=price,
                        status=OrderStatus.rejected,
                        error=translate_error_message(code="rate_limited"),
                    )
                    failures += 1
                    continue

                try:
                    enforce_whitelists(
                        connector=connector_name,
                        symbol=symbol,
                        order_type=order_type,
                    )
                except RiskError as exc:
                    await self._record_exec_order(
                        session=session,
                        signal=signal,
                        user=user,
                        account_name=account_name,
                        connector=connector_name,
                        symbol=symbol,
                        side=trade_type,
                        order_type=order_type,
                        position_action=position_action,
                        qty=Decimal("0"),
                        price=price,
                        status=OrderStatus.rejected,
                        error=translate_error_message(code=exc.code),
                    )
                    failures += 1
                    continue

                cache_key = (connector_name.lower(), symbol.upper())
                rule = rules_cache.get(cache_key)
                if rule is None:
                    try:
                        rule = await hb_client.get_trading_rules(connector_name, symbol)
                    except HBClientError as exc:
                        logger.error(
                            "Failed to fetch trading rules from hummingbot-api for %s/%s: %s",
                            connector_name,
                            symbol,
                            exc,
                        )
                        await self._append_audit(
                            session=session,
                            user_id=None,
                            action="signal.error",
                            target=f"script:{script.id}",
                            payload={
                                "signal_id": signal.id,
                                "connector": connector_name,
                                "error": "hb_unreachable",
                                "detail": str(exc),
                            },
                        )
                        await self._record_exec_order(
                            session=session,
                            signal=signal,
                            user=user,
                            account_name=account_name,
                            connector=connector_name,
                            symbol=symbol,
                            side=trade_type,
                            order_type=order_type,
                            position_action=position_action,
                        qty=Decimal("0"),
                        price=price,
                        status=OrderStatus.error,
                        error=translate_error_message(code="hb_unreachable"),
                    )
                        failures += 1
                        continue
                    rules_cache[cache_key] = rule

                # 若 trading_pair 与脚本 symbol 不一致则忽略（记录 audit 不下单）
                if symbol.upper() != (script.symbol or "").upper():
                    await self._append_audit(
                        session=session,
                        user_id=user.id,
                        action="signal.ignored.symbol_mismatch",
                        target=f"script:{script.id}",
                        payload={
                            "signal_id": signal.id,
                            "expect": script.symbol,
                            "got": symbol,
                        },
                    )
                    continue

                try:
                    exec_result = await self._execute_order(
                        session=session,
                        user=user,
                        subscription=sub,
                        script=script,
                        signal=signal,
                        rule=rule,
                        trade_type=trade_type,
                        order_type=order_type,
                        position_action=position_action,
                        symbol=symbol,
                        price=price,
                        account_name=account_name,
                        connector_name=connector_name,
                        order_book_cache=order_book_cache,
                    )
                except RiskError as exc:
                    await self._record_exec_order(
                        session=session,
                        signal=signal,
                        user=user,
                        account_name=account_name,
                        connector=connector_name,
                        symbol=symbol,
                        side=trade_type,
                        order_type=order_type,
                        position_action=position_action,
                        qty=Decimal("0"),
                        price=price,
                        status=OrderStatus.rejected,
                        error=translate_error_message(code=exc.code),
                    )
                    failures += 1
                    continue
                except Exception as exc:  # noqa: BLE001
                    await self._record_exec_order(
                        session=session,
                        signal=signal,
                        user=user,
                        account_name=account_name,
                        connector=connector_name,
                        symbol=symbol,
                        side=trade_type,
                        order_type=order_type,
                        position_action=position_action,
                        qty=Decimal("0"),
                        price=price,
                        status=OrderStatus.error,
                        error=translate_error_message(raw=str(exc)),
                    )
                    failures += 1
                    continue

                if exec_result.status in {OrderStatus.filled, OrderStatus.sent, OrderStatus.partial}:
                    success += 1
                else:
                    failures += 1

            if success and failures:
                final_status = SignalStatus.partial
                error = None
            elif success and not failures:
                final_status = SignalStatus.filled
                error = None
            else:
                final_status = SignalStatus.error
                error = translate_error_message(code="all_failed")

            signal.status = final_status
            signal.error = error
            signal.processed_at = now_utc()
            await session.commit()

    async def _mark_signal(
        self,
        session: AsyncSession,
        signal: ScriptSignal,
        status: SignalStatus,
        error: Optional[str] = None,
    ) -> None:
        signal.status = status
        signal.error = error
        signal.processed_at = now_utc()
        await session.commit()

    async def _fetch_subscriptions(
        self,
        session: AsyncSession,
        script_id: int,
    ) -> List[Tuple[ScriptSubscription, User]]:
        stmt = (
            select(ScriptSubscription, User)
            .join(User, ScriptSubscription.user_id == User.id)
            .where(ScriptSubscription.script_id == script_id)
        )
        result = await session.execute(stmt)
        rows = result.all()
        return [(row[0], row[1]) for row in rows]

    async def _execute_order(
        self,
        session: AsyncSession,
        user: User,
        subscription: ScriptSubscription,
        script: Script,
        signal: ScriptSignal,
        rule: Dict,
        trade_type: str,
        order_type: str,
        position_action: str,
        symbol: str,
        price: Optional[Decimal],
        account_name: str,
        connector_name: str,
        order_book_cache: Dict[Tuple[str, str], Tuple[Optional[Decimal], Optional[Decimal]]],
    ) -> ExecOrder:
        allocation = subscription.allocation or {}
        target = signal.payload_json.get("target") or {}

        invest_value: Optional[Decimal] = None
        if isinstance(target, dict) and target.get("value") is not None:
            invest_value = Decimal(str(target.get("value")))
        elif isinstance(allocation, dict) and allocation.get("value") is not None:
            invest_value = Decimal(str(allocation.get("value")))
        if invest_value is None:
            raise RiskError("ERR_ALLOCATION", "Missing investment ratio")

        if invest_value <= 0:
            raise RiskError("ERR_ALLOCATION", "Investment ratio must be positive")

        if invest_value > 1:
            invest_ratio = invest_value / Decimal("100")
        else:
            invest_ratio = invest_value

        equity_value = await self._fetch_equity(account_name)
        if equity_value <= 0:
            raise RiskError("ERR_EQUITY", "Account equity unavailable")
        notional_value = Decimal(str(equity_value)) * invest_ratio
        if subscription.max_notional:
            enforce_max_notional(notional_value, Decimal(subscription.max_notional))
        else:
            enforce_max_notional(notional_value, None)

        # 交易所最小名义额（以报价币计）预检，避免把无效单提交到 hb-api
        try:
            from .risk import _decimal  # type: ignore
        except Exception:  # noqa: BLE001
            _decimal = None  # type: ignore
        if _decimal is not None:
            min_notional_rule = _decimal(
                rule,
                ("min_notional", "min_order_value", "min_quote_amount", "min_base_amount"),
                Decimal("0"),
            )
            if notional_value < min_notional_rule:
                # 写审计并返回更友好的错误
                await self._append_audit(
                    session=session,
                    user_id=user.id,
                    action="signal.rejected.min_notional",
                    target=f"script:{script.id}",
                    payload={
                        "signal_id": signal.id,
                        "connector": connector_name,
                        "symbol": symbol,
                        "min_notional": str(min_notional_rule),
                        "notional": str(notional_value),
                    },
                )
                raise RiskError(
                    "ERR_MIN_NOTIONAL",
                    f"名义金额不足：需要 ≥ {min_notional_rule}（报价币），当前 {notional_value}"
                )

        price_decimal: Optional[Decimal] = None
        if price is not None:
            price_decimal = Decimal(str(price))

        slippage_bps = subscription.slippage_bps
        if slippage_bps is None:
            slippage_bps = settings.slippage_bps_limit

        if order_type.upper() == "MARKET":
            best_bid, best_ask = await self._get_best_quotes(connector_name, symbol, order_book_cache)
        else:
            best_bid, best_ask = (None, None)

        normalized_qty, normalized_price = self._compute_amount_and_price(
            market_type=script.market_type,
            order_type=order_type,
            trade_type=trade_type,
            position_action=position_action,
            notional=notional_value,
            explicit_price=price_decimal,
            rule=rule,
            best_bid=best_bid,
            best_ask=best_ask,
            slippage_bps=int(slippage_bps),
        )

        exec_order = ExecOrder(
            signal_id=signal.id,
            user_id=user.id,
            account_name=account_name,
            connector=connector_name,
            symbol=symbol,
            side=TradeSide(trade_type),
            order_type=order_type,
            position_action=PositionAction(position_action),
            qty=normalized_qty,
            price=normalized_price,
            status=OrderStatus.sent,
        )
        session.add(exec_order)
        await session.flush()

        # hb-api 对 amount/price 期望为数值类型；
        # 市价单通常要求 price=0，这里做兼容：LIMIT 用归一化后的价格，MARKET 设置为 0。
        payload = {
            "account_name": exec_order.account_name,
            "connector_name": connector_name,
            "trading_pair": symbol,
            "trade_type": trade_type,
            "order_type": order_type,
            "position_action": position_action,
            "amount": float(normalized_qty),
            "price": float(normalized_price) if (normalized_price is not None) else 0.0,
        }

        try:
            response = await hb_client.place_order(payload)
        except Exception as exc:  # noqa: BLE001
            exec_order.status = OrderStatus.error
            exec_order.error = translate_error_message(raw=str(exc))
            await self._append_audit(
                session=session,
                user_id=user.id,
                action="order.place.error",
                target=f"script:{script.id}",
                payload={
                    "signal_id": signal.id,
                    "connector": connector_name,
                    "payload": payload,
                    "error": exec_order.error,
                },
            )
            await session.commit()
            await session.refresh(exec_order)
            return exec_order

        exec_order.tx_ref = response.get("order_id") or response.get("id")
        remote_status = response.get("status", "").lower()
        if remote_status in {"filled", "completed"}:
            exec_order.status = OrderStatus.filled
        elif remote_status in {"partial", "partially_filled"}:
            exec_order.status = OrderStatus.partial
        else:
            exec_order.status = OrderStatus.sent
        response_error = response.get("error") if isinstance(response, dict) else None
        if response_error and exec_order.status == OrderStatus.error:
            exec_order.error = translate_error_message(raw=response_error)

        await self._append_audit(
            session=session,
            user_id=user.id,
            action="order.place",
            target=f"script:{script.id}",
            payload={
                "signal_id": signal.id,
                "payload": payload,
                "response": response,
            },
        )

        await session.commit()
        await session.refresh(exec_order)
        return exec_order

    async def _fetch_equity(self, account_name: str) -> float:
        try:
            portfolio = await hb_client.get_portfolio_state(account_name)
        except Exception:  # noqa: BLE001
            return 0.0
        if not portfolio:
            return 0.0
        return float(
            portfolio.get("total_equity")
            or portfolio.get("portfolio_value")
            or portfolio.get("balances", {}).get("equity", 0.0)
        )

    async def _record_exec_order(
        self,
        session: AsyncSession,
        signal: ScriptSignal,
        user: Optional[User],
        account_name: str,
        connector: str,
        symbol: str,
        side: str,
        order_type: str,
        position_action: str,
        qty: Decimal,
        price: Optional[Decimal],
        status: OrderStatus,
        error: Optional[str],
    ) -> ExecOrder:
        exec_order = ExecOrder(
            signal_id=signal.id,
            user_id=user.id if user else None,
            account_name=account_name,
            connector=connector,
            symbol=symbol,
            side=TradeSide(side),
            order_type=order_type,
            position_action=PositionAction(position_action),
            qty=qty,
            price=price,
            status=status,
            error=error,
        )
        session.add(exec_order)
        await session.flush()
        await session.commit()
        return exec_order

    async def _append_audit(
        self,
        session: AsyncSession,
        user_id: Optional[int],
        action: str,
        target: str,
        payload: Dict,
    ) -> None:
        log = AuditLog(
            actor_id=user_id,
            action=action,
            target=target,
            payload_json=payload,
        )
        session.add(log)
        await session.flush()


signal_worker = SignalWorker(concurrency=get_settings().worker_concurrency)
